package com.ls.fw.search.solrj.dao.base;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import net.sf.json.JSONObject;

import org.apache.http.client.HttpClient;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.ConcurrentUpdateSolrClient;
import org.apache.solr.client.solrj.impl.HttpClientUtil;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
import org.apache.solr.client.solrj.response.UpdateResponse;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;

import com.ls.fw.data.exception.DataException;

public class BaseSolrDaoImpl implements BaseSolrDao{

	public static final String DEFAULT_NAME = "collection1";
	public static final String DEFAULT_COLLECTION = "collection1";
	protected static final Map<String,String> schemaPathMap = new HashMap<String,String>(); 
	
	private SolrClient server = null;// solr服务
	
	private String baseURL = "";//http://127.0.0.1/solr/；或者127.0.0.1\:9983
	
	private String name = DEFAULT_NAME;//名称
	
	private boolean autoCommit = false;//是否自动提交
	
	private boolean cloud = false;//是否为分布式
	
	
	public BaseSolrDaoImpl() {
		this(null,DEFAULT_NAME);
	}
	
	public BaseSolrDaoImpl(String baseUrl) {
		this(baseUrl,DEFAULT_NAME);
	}
	
	public BaseSolrDaoImpl(SolrClient server) {
		this.server = server;
	}
	
	public BaseSolrDaoImpl(String baseURL, String name) {
		super();
		this.baseURL = baseURL;
		this.name = name;
		if(baseURL!=null){
			this.server = this.newSolrClient(this.baseURL, this.name);
		}
	}
	
	@Override
	public UpdateResponse add(Map<String, Object> data) {
		if(data==null){
			throw new DataException("索引数据不能为空！");
		}
		SolrInputDocument doc = new SolrInputDocument();
		boolean existId = false;
		for(String field : data.keySet()){
			if(field.equals(DEFAULT_PK)){
				existId = true;
			}
			doc.addField(field, data.get(field));
		}
		if(!existId){
			doc.addField(DEFAULT_PK, UUID.randomUUID().toString());
		}
		UpdateResponse res = null;
		try {
			res = this.server.add(doc);
			if(this.autoCommit){
				this.commit();
			}
		} catch (Exception e) {
			if(this.autoCommit){
				this.rollback();
			}
			throw new DataException("添加索引异常！",e);
		}finally{
		}
		return res;
	}
	
	
	@Override
	public UpdateResponse add(Object model) {
		if(model==null){
			throw new DataException("索引数据不能为空！");
		}
		Map<String, Object> data = null;
		//TODO:反射对象
		
		return this.add(data);
	}
	
	
	public UpdateResponse save(Object model) {
		return this.add(model);
	}
	
	
	@Override
	public SolrClient buildSolrClient(String baseURL, String name){
		if(this.baseURL.equals(baseURL) && this.name.equals(name)){
			if(this.server!=null){
				return this.server;
			}
		}
		return this.newSolrClient(baseURL,name);
	}
	
	
	@Override
	public SolrClient newSolrClient(String baseUrl, String name){
		if(!baseUrl.startsWith("http:")){
			this.cloud = true;
		}
		if(this.cloud){
			return  this.getCloudSolrClient(baseUrl, name);
		}else{
			return this.getHttpSolrClient(baseUrl);
		}
	}
	
	/**
	 * 该类实例化多用于更新删除索引操作
	 * @param solrServerUrl 是Solr服务器的地址，
	 * @param queueSize 缓冲区大小
	 * @param threadCount 后台线程数
	 * @return
	 * @author ls
	 * 2014年8月19日下午9:34:24
	 */
	protected ConcurrentUpdateSolrClient getConcurrentUpdateSolrClient(String solrServerUrl,  int queueSize,  int threadCount){
		return new ConcurrentUpdateSolrClient(solrServerUrl,queueSize,threadCount);
	}
	
	
	/**
	 * 用于有多个Solr服务器，负载均衡
	 * @param solrServerUrl
	 * @param queueSize
	 * @param threadCount
	 * @return
	 * @author ls
	 * 2014年8月19日下午10:19:19
	 * @throws MalformedURLException 
	 */
	protected LBHttpSolrClient getLBHttpSolrClient(String... solrServerUrls) throws MalformedURLException{
		return new LBHttpSolrClient(solrServerUrls);
	}
	

	/**
	 * 该类实例化多用于查询操作
	 * @param baseUrl
	 * @return
	 * @author ls
	 * 2014年8月19日下午9:35:56
	 */
	protected HttpSolrClient getHttpSolrClient(String baseUrl){
		ModifiableSolrParams params = new ModifiableSolrParams();
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS, 128);
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, 32);
		//params.set(HttpClientUtil.PROP_FOLLOW_REDIRECTS, followRedirects);
		HttpClient httpClient =  HttpClientUtil.createClient(params);
		//该类实例化多用于查询操作
		return new HttpSolrClient(baseUrl+"/"+name,httpClient);

	}

	protected CloudSolrClient getCloudSolrClient(String zkHostString,
			String collection) {
		ModifiableSolrParams params = new ModifiableSolrParams();
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS, 1000);//10
		params.set(HttpClientUtil.PROP_MAX_CONNECTIONS_PER_HOST, 500);//5
		HttpClient client = HttpClientUtil.createClient(params);						
		LBHttpSolrClient lbServer = new LBHttpSolrClient(client);

		CloudSolrClient server1 = null;
		server1 = new CloudSolrClient(zkHostString, lbServer, true);
		server1.setZkClientTimeout(30000);
		server1.setZkConnectTimeout(30000);
		if (collection != null) {
			server1.setDefaultCollection(collection);
		}
		server1.setParallelUpdates(true);
		server1.connect();
		return server1;
	}
	
	@Override
	public SolrClient switchName(String name){
		this.name = name;
		return this.reopen();
	}
	
	
	@Override
	public SolrClient switchServer(String baseUrl,String name){
		this.baseURL = baseUrl;
		return this.switchName(name);
	}
	
	
	@Override
	public SolrClient switchServer(String baseUrl){
		this.baseURL = baseUrl;
		return this.switchName(DEFAULT_NAME);
	}
	
	@Override
	public void commit()  {
		try {
			if(server!=null){
				this.server.commit();
			}
		} catch (Exception e) {
			throw new DataException(e.getMessage(),e);
		}
	}
	
	@Override
	public void rollback()  {
		try {
			if(server!=null){
				this.server.rollback();
			}
		} catch (Exception e) {
			throw new DataException(e.getMessage(),e);
		}
	}
	
	

	@Override
	public SolrClient open(){
		synchronized (this) {
			if(this.server==null){
				this.server = this.newSolrClient(baseURL, name);
			}
		}
		return server;
	}
	
	@Override
	public SolrClient reopen(){
		this.close();
		return this.open();
	}
	
	@Override
	public void close() {
		try{
			this.commit();
		}catch (Exception e) {
			e.printStackTrace();
		}
		try {
			if(server!=null){
				if(this.cloud){
					  ZkStateReader reader = ((CloudSolrClient)server).getZkStateReader();
					  if(reader!=null){
					      reader.close();
					  }
				}
				this.server.close();
			}
			server = null;
		} catch (Exception e) {
			throw new DataException(e.getMessage(),e);
		}
	}
	
	@Override
	public SolrClient getServer() {
		return server;
	}
	
	public void setServer(SolrClient server) {
		this.server = server;
	}
	
	@Override
	public String getBaseURL() {
		return baseURL;
	}
	
	
	public void setBaseURL(String baseURL) {
		this.baseURL = baseURL;
	}
	
	@Override
	public String getName() {
		return name;
	}
	
	@Override
	public boolean isAutoCommit() {
		return autoCommit;
	}

	private String getSchemaPath(){
		String schemaPath = schemaPathMap.get(this.getName());
		if(schemaPath==null || "".equals(schemaPath)){
			BufferedReader br = null;
			try {
				String schemaPathUrl = this.getBaseURL()+"/admin/cores?wt=json&action=SCHEMAPATH&core="+this.getName()+"&_="+new Date().getTime();
		        URL url = new URL(schemaPathUrl);
		        HttpURLConnection urlConnection = (HttpURLConnection) url.openConnection();
		        InputStream is = urlConnection.getInputStream();  
		        br = new BufferedReader(new InputStreamReader(is));  
		        StringBuilder sb = new StringBuilder();  
		        String str = null;
		        while((str = br.readLine())!=null){  
		            sb.append(str);  
		        }  
		        String content = new String(sb);
		        content = new String(content.getBytes("UTF-8"));
		        JSONObject json = JSONObject.fromObject(content);
		        schemaPath = json.getString("schemePath");
		        JSONObject error = json.getJSONObject("error");
		        json = json.getJSONObject("responseHeader");
		        int status = json.getInt("status");
		        if(status!=0&&error!=null){
		        	throw new DataException(error.getString("msg"));
		        }else{
		        	schemaPathMap.put(this.getName(), schemaPath);
		        }
			} catch (DataException e) {
		        throw e;
	        } catch (Exception e) {
		        throw new DataException("刷新索引异常!",e);
	        } finally{
	        	if(br!=null){
	        		try {
						br.close();
					} catch (IOException e) {
						e.printStackTrace();
					}  
	        	}
	        }
		}
		
		return schemaPath;
	}

	  

	@Override
	public boolean isCloud() {
		return this.cloud;
	}
	
	
	public BaseSolrDao cloud(boolean cloud){
		this.cloud = cloud;
		return this;
	}
	

	
	public BaseSolrDao name(String name) {
		this.name = name;
		return this;
	}

	public BaseSolrDao autoCommit(boolean autoCommit) {
		this.autoCommit = autoCommit;
		return this;
	}
	
}
